/*
Copyright [2020] [https://www.xiaonuo.vip]

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

  http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

Snowy采用APACHE LICENSE 2.0开源协议，您在使用过程中，需要注意以下几点：

1.请不要删除和修改根目录下的LICENSE文件。
2.请不要删除和修改Snowy源码头部的版权声明。
3.请保留源码和相关描述文件的项目出处，作者声明等。
4.分发源码时候，请注明软件出处 https://gitee.com/xiaonuobase/snowy
5.在修改包名，模块名称，项目代码等时，请注明软件出处 https://gitee.com/xiaonuobase/snowy
6.若您的项目无法满足以上几点，可申请商业授权，获取Snowy商业授权许可，请在官网购买授权，地址为 https://www.xiaonuo.vip
 */
package vip.xiaonuo.modular.manage.tsdatarecord.service.impl;

import cc.blynk.clickhouse.ClickHouseConnection;
import cc.blynk.clickhouse.ClickHouseDataSource;
import cn.afterturn.easypoi.excel.entity.params.ExcelExportEntity;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import vip.xiaonuo.core.dbs.CurrentDataSourceContext;
import vip.xiaonuo.core.exception.ServiceException;
import vip.xiaonuo.core.factory.PageFactory;
import vip.xiaonuo.core.pojo.page.PageResult;
import vip.xiaonuo.core.util.PoiUtil;
import vip.xiaonuo.dbs.core.annotation.DataSource;
import vip.xiaonuo.modular.common.DistanceHelper;
import vip.xiaonuo.modular.common.LoadClickhouseDatasource;
import vip.xiaonuo.modular.common.WdAnalysisConst;
import vip.xiaonuo.modular.gaode.Address;
import vip.xiaonuo.modular.gaode.CoordinateConvertAddr;
import vip.xiaonuo.modular.manage.biindexdata.service.BiIndexDataService;
import vip.xiaonuo.modular.manage.biindextype.service.BaseDataTitleRelevanceService;
import vip.xiaonuo.modular.cache.BaseDataRecordCache;
import vip.xiaonuo.modular.manage.tsdatarecord.entity.*;
import vip.xiaonuo.modular.manage.tsdatarecord.enums.FileParseRecordStatusEnum;
import vip.xiaonuo.modular.manage.tsdatarecord.enums.TimeSpaceDataRecordExceptionEnum;
import vip.xiaonuo.modular.manage.tsdatarecord.mapper.BaseDataRecordMapper;
import vip.xiaonuo.modular.manage.tsdatarecord.param.DataParam;
import vip.xiaonuo.modular.manage.tsdatarecord.param.ExportParam;
import vip.xiaonuo.modular.manage.tsdatarecord.param.TimeSpaceDataRecordParam;
import vip.xiaonuo.modular.manage.tsdatarecord.parser.FileParser;
import vip.xiaonuo.modular.manage.tsdatarecord.parser.FileParserSupport;
import vip.xiaonuo.modular.manage.tsdatarecord.parser.ParserResult;
import vip.xiaonuo.modular.manage.tsdatarecord.service.BaseDataDefineService;
import vip.xiaonuo.modular.manage.tsdatarecord.service.BaseDataRecordService;
import vip.xiaonuo.modular.manage.tsdatarecord.service.BaseDataTableService;
import vip.xiaonuo.modular.manage.tsdatarecord.service.BaseDataTypeService;

import javax.annotation.Resource;
import java.awt.*;
import java.io.File;
import java.math.BigDecimal;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.MessageFormat;
import java.util.List;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 时空数据service接口实现类
 *
 * @author cy
 * @date 2022-07-28 19:37:28
 */
@Service
@Slf4j
public class BaseDataRecordServiceImpl extends ServiceImpl<BaseDataRecordMapper, BaseDataRecord> implements BaseDataRecordService {

    public static final String CH_TABLE_PREFIX = "BASE_DATA_";

    private static final int PAGE_SIZE = 100;

    @Autowired
    private BaseDataTableService baseDataTableService;
    @Autowired
    private BaseDataTitleRelevanceService titleRelevanceService;
    @Autowired
    private BiIndexDataService biIndexDataService;
    @Autowired
    private BaseDataDefineService baseDataDefineService;
    @Autowired
    private BaseDataTypeService baseDataTypeService;


    @Override
    public void defineData(BaseDataDefine record) {
        BaseDataRecord dataRecord = this.getById(record.getRecordId());
        if (ObjectUtil.isNull(dataRecord))
            throw new ServiceException(TimeSpaceDataRecordExceptionEnum.NOT_EXIST);
        //更新数据类型
        dataRecord.setTypeCode(record.getTypeCode());
        //更新数据状态
        dataRecord.setStatus(FileParseRecordStatusEnum.IMPORTING.getStatus());

        //获取该数据类型的相关信息
        BaseDataType dataType = baseDataTypeService.getByCode(record.getTypeCode());
        if (ObjectUtil.isNull(dataType))
            throw new ServiceException(TimeSpaceDataRecordExceptionEnum.NOT_EXIST);

        //查询表格属性
        List<BaseDataTable> baseDataTables = baseDataTableService.findListByRecordId(record.getRecordId());
        //构建该类型数据的ClickHouse表
        this.buildTable(baseDataTables, dataRecord, dataType);

        //更新base_data_define
        BaseDataDefine dataDefine = baseDataDefineService.getByRecordId(record.getRecordId());
        if (ObjectUtil.isNull(dataDefine)) {
            baseDataDefineService.save(record);
        } else {
            BeanUtil.copyProperties(record, dataDefine, "id", "createTime", "createUser");
            baseDataDefineService.updateById(dataDefine);
        }
        //更新base_data_record
        this.updateById(dataRecord);

        //开启线程去执行数据导入
        ThreadUtil.execute(() -> importData(record));
    }

    /**
     * 从 base_matedata导数据到各个类型的表中
     *
     * @param record
     */
    private void importData(BaseDataDefine record) {
        Long recordId = record.getRecordId();
        BaseDataRecord dataRecord = this.getById(recordId);
        List<BaseDataTable> baseDataTables = baseDataTableService.findListByRecordId(recordId);
        String insertSql = strSql(baseDataTables, record);
        //设置经纬度对应的字段
        setLngLatDataIndex(record, baseDataTables);

        if (record.getIsColumnTime()) {

        } else {
            BaseDataDefine.TimeWrapper timeWrapper = buildTimeWrapper(record);
            ClickHouseConnection connection = null;
            try {

                ClickHouseDataSource clickHouseDataSource = LoadClickhouseDatasource.get();
                connection = clickHouseDataSource.getConnection();
                connection.setAutoCommit(false);
                //切换clickhouse数据源
                CurrentDataSourceContext.setDataSourceType(WdAnalysisConst.CLICKHOUSE_DATASOURCE);
                //先根据recordId删除数据
                this.getBaseMapper().deleteBaseData(CH_TABLE_PREFIX + record.getTypeCode(), recordId.toString());
                int pageNo = 1;
                int totalPage = 1;
                while (pageNo < totalPage + 1) {
                    PageResult<BaseMetadata> metadataPage = pageMetaData(recordId.toString(), pageNo, PAGE_SIZE);
                    Map<String, DefaultCommonData> defaultCommonDataMap = new HashMap<>();
                    //解析数据
                    List<BaseMetadata> metadataList = metadataPage.getRows();
                    for (BaseMetadata metadata : metadataList) {
                        DefaultCommonData commonData = new DefaultCommonData();
                        commonData.setId(IdWorker.getId());
                        //set公共字段
                        commonData.setRecordId(recordId.toString());
                        commonData.setTypeCode(record.getTypeCode());
                        //set年月日
                        commonData.setYear(timeWrapper.getYear());
                        commonData.setMonth(timeWrapper.getMonth());
                        commonData.setDay(timeWrapper.getDay());

                        JSONObject jsonObject = JSON.parseObject(metadata.getContent());
                        //获取坐标
                        String location = getLngLat(record, jsonObject, commonData);
                        //set不同类型的具体数据
                        commonData.setIndexDataMap(jsonObject.getInnerMap());
                        if (StringUtils.isNotBlank(location)) {
                            defaultCommonDataMap.put(location, commonData);
                        }
                    }
                    //批量获取省市县
                    Map<String, Address> addressMap = CoordinateConvertAddr.getBatchLocationAddrMap(new ArrayList<>(defaultCommonDataMap.keySet()));
                    addressMap.entrySet().forEach(entry -> {
                        Address address = entry.getValue();
                        if (defaultCommonDataMap.containsKey(entry.getKey())) {
                            DefaultCommonData cd = defaultCommonDataMap.get(entry.getKey());
                            cd.setProvince(address.getProvName());
                            cd.setCity(address.getCityName());
                            cd.setRegion(address.getCountryName());
                        }
                    });

                    List<DefaultCommonData> dataList = new ArrayList<>(defaultCommonDataMap.values());
                    saveDefaultCommonData(dataList, baseDataTables, connection, insertSql);
                    totalPage = metadataPage.getTotalPage();
                    log.info("总共{}条 {}页数据，导入第{}页{}条数据完成！", metadataPage.getTotalRows(), totalPage, pageNo, dataList.size());
                    pageNo++;
                }
                CurrentDataSourceContext.clearDataSourceType();

                dataRecord.setStatus(FileParseRecordStatusEnum.IMPORTED.getStatus());
                this.updateById(dataRecord);
            } catch (Exception e) {
                log.error(">>>base_data_record:{} 导入数据失败！", recordId);
                log.error("导入数据失败！", e);
                CurrentDataSourceContext.clearDataSourceType();
                dataRecord.setStatus(FileParseRecordStatusEnum.IMPORT_FAIL.getStatus());
                this.updateById(dataRecord);
            } finally {
                if (connection != null) {
                    try {
                        connection.close();
                    } catch (SQLException ignore) {
                    }
                }
            }
        }
    }


    private String getLngLat(BaseDataDefine record, JSONObject jsonObject, DefaultCommonData commonData) {
        String lngLat = null;
        if (record.getType() == 0) {// 点
            String lngDataIndex = record.getLngDataIndex();
            String latDataIndex = record.getLatDataIndex();
            if (!StringUtils.isAnyBlank(lngDataIndex, latDataIndex)) {
                String lngStr = StringUtils.trimToEmpty(jsonObject.getString(lngDataIndex));
                String latStr = StringUtils.trimToEmpty(jsonObject.getString(latDataIndex));
                if (NumberUtil.isDouble(lngStr) && NumberUtil.isDouble(latStr)) {
                    commonData.setLng(lngStr);
                    commonData.setLat(latStr);
                    commonData.setMultipolygon("");
                    lngLat = lngStr + "," + latStr;
                } else {
                    log.error(">>> {}指定的经纬度坐标不是数字: {}", record.getRecordId(), lngStr + "," + latStr);
                }
            }
        } else {// the_geom
            //多边形面 MULTIPOLYGON (((114.39284688781275 30.668386856649345, 114.38237688781278 30.668386856649345, 114.38237688781278 30.67743685664941, 114.39284688781275 30.67743685664941, 114.39284688781275 30.668386856649345)))
            //点  POINT (114.289335856 30.5768343128)
            String coverDataIndex = record.getCoverDataIndex();
            if (StringUtils.isNotBlank(coverDataIndex)) {
                String coverStr = StringUtils.trimToEmpty(jsonObject.getString(coverDataIndex));
                if (coverStr.startsWith("MULTIPOLYGON")) {
                    String[] coverArr = coverStr.split("\\(\\(\\(");
                    if (coverArr.length == 2) {
                        String multipolygon = coverArr[1].replace(")))", "");
                        String[] splitloc = multipolygon.split(",");

                        String[] lngLats = splitloc[splitloc.length - 1].trim().split(" ");
                        if (lngLats.length == 2) {
                            lngLat = StringUtils.join(lngLats, ",");
                            commonData.setLng(lngLats[0]);
                            commonData.setLat(lngLats[1]);
                        } else {
                            lngLat = "";
                            commonData.setLng("");
                            commonData.setLat("");
                        }
                        commonData.setMultipolygon(multipolygon);
                    } else {
                        log.error(">>> {}指定的多边形坐标格式不正确: {}", record.getRecordId(), coverStr);
                    }
                }
                if (coverStr.startsWith("POINT")) {
                    String[] coverArr = coverStr.split("\\(");
                    if (coverArr.length == 2) {
                        String point = coverArr[1].replace(")", "");
                        String[] splitloc = point.split(" ");
                        lngLat = splitloc[0] + "," + splitloc[1];
                        commonData.setLng(splitloc[0]);
                        commonData.setLat(splitloc[1]);
                        commonData.setMultipolygon("");
                    } else {
                        log.error(">>> {}指定的点坐标格式不正确: {}", record.getRecordId(), coverStr);
                    }
                }

            }
        }
        return lngLat;
    }

    /**
     * 逆地理编码 获取省市区
     *
     * @param record
     * @param jsonObject
     * @param commonData
     */
    private void regeo(BaseDataDefine record, JSONObject jsonObject, DefaultCommonData commonData) {
        if (record.getType() == 0) {// 点
            String lngDataIndex = record.getLngDataIndex();
            String latDataIndex = record.getLatDataIndex();
            if (!StringUtils.isAnyBlank(lngDataIndex, latDataIndex)) {
                String lngStr = StringUtils.trimToEmpty(jsonObject.getString(lngDataIndex));
                String latStr = StringUtils.trimToEmpty(jsonObject.getString(latDataIndex));
                if (NumberUtil.isDouble(lngStr) && NumberUtil.isDouble(latStr)) {
                    String[] locationAddr = CoordinateConvertAddr.getBatchLocationAddr(lngStr + "," + latStr);
                    commonData.setLng(lngStr);
                    commonData.setLat(latStr);
                    commonData.setMultipolygon("");
                    if (locationAddr.length == 5) {
                        commonData.setProvince(locationAddr[2]);
                        commonData.setCity(locationAddr[3]);
                        commonData.setRegion(locationAddr[4]);
                    } else {
                        commonData.setProvince("");
                        commonData.setCity("");
                        commonData.setRegion("");
                    }
                } else {
                    log.error(">>> {}指定的经纬度坐标不是数字: {}", record.getRecordId(), lngStr + "," + latStr);
                }
            }
        } else {// 面
            // MULTIPOLYGON (((114.39284688781275 30.668386856649345, 114.38237688781278 30.668386856649345, 114.38237688781278 30.67743685664941, 114.39284688781275 30.67743685664941, 114.39284688781275 30.668386856649345)))
            String coverDataIndex = record.getCoverDataIndex();
            if (StringUtils.isNotBlank(coverDataIndex)) {
                String coverStr = StringUtils.trimToEmpty(jsonObject.getString(coverDataIndex));
                String[] coverArr = coverStr.split("\\(\\(\\(");
                if (coverArr.length == 2) {
                    String multipolygon = coverArr[1].replace(")))", "");
                    String[] splitloc = multipolygon.split(",");
                    String location = splitloc[splitloc.length - 1].trim().replace(" ", ",");
                    String[] locationAddr = CoordinateConvertAddr.getBatchLocationAddr(location);
                    commonData.setLng("");
                    commonData.setLat("");
                    commonData.setMultipolygon(multipolygon);
                    if (locationAddr.length == 5) {
                        commonData.setProvince(locationAddr[2]);
                        commonData.setCity(locationAddr[3]);
                        commonData.setRegion(locationAddr[4]);
                    } else {
                        commonData.setProvince("");
                        commonData.setCity("");
                        commonData.setRegion("");
                    }
                } else {
                    log.error(">>> {}指定的多边形坐标格式不正确: {}", record.getRecordId(), coverStr);
                }
            }
        }
    }

    /**
     * 获取经纬度对应的字段下标
     *
     * @param record
     * @param baseDataTables
     * @return
     */
    private int setLngLatDataIndex(BaseDataDefine record, List<BaseDataTable> baseDataTables) {
        int type = record.getType();//0:点  1:面
        if (type == 0) {
            Optional<BaseDataTable> lngTableColumnOptional = baseDataTables.stream().filter(b -> b.getId().equals(record.getLngColumnTableId())).findFirst();
            record.setLngDataIndex(lngTableColumnOptional.isPresent() ? lngTableColumnOptional.get().getDataIndex() : "");
            Optional<BaseDataTable> latTableColumnOptional = baseDataTables.stream().filter(b -> b.getId().equals(record.getLatColumnTableId())).findFirst();
            record.setLatDataIndex(latTableColumnOptional.isPresent() ? latTableColumnOptional.get().getDataIndex() : "");
            if (StringUtils.isAnyBlank(record.getLngDataIndex(), record.getLatDataIndex())) {
                log.error(">>>{} 指定的经纬度字段为空", record.getRecordId());
            }
        } else {
            Optional<BaseDataTable> coverTableColumnOptional = baseDataTables.stream().filter(b -> b.getId().equals(record.getCoverColumnTableId())).findFirst();
            record.setCoverDataIndex(coverTableColumnOptional.isPresent() ? coverTableColumnOptional.get().getDataIndex() : "");
            if (StringUtils.isBlank(record.getCoverDataIndex())) {
                log.error(">>>{} 指定的多边形坐标字段为空", record.getRecordId());
            }
        }
        return type;
    }

    /**
     * 分页查询base_matedata
     *
     * @param recordId
     * @param pageNo
     * @param pageSize
     * @return
     */
    private PageResult<BaseMetadata> pageMetaData(String recordId, int pageNo, int pageSize) {
        return new PageResult<>(this.getBaseMapper().selectMetadataPage(new Page<>(pageNo, pageSize), recordId));
    }

    /**
     * 生成insert sql
     *
     * @param record
     * @return
     */
    private static final MessageFormat default_sql_format = new MessageFormat("insert into {0}" +
            "(id , record_id , type_code , `year` , `month` , `day` , province , city , region , lng , lat , multipolygon, " +
            "{1}) values " +
            "(? , ? , ? , ? , ? , ? , ? , ? , ? , ? , ? , ? , {2})");

    private String strSql(List<BaseDataTable> baseDataTables, BaseDataDefine record) {
        List<String> columnList = baseDataTables.stream().map(BaseDataTable::getDataIndex).collect(Collectors.toList());
        List<String> valueList = baseDataTables.stream().map(b -> "?").collect(Collectors.toList());

        String tableName = CH_TABLE_PREFIX + record.getTypeCode();
        String columns = StringUtils.join(columnList, ",");
        String values = StringUtils.join(valueList, ",");
        return default_sql_format.format(new String[]{tableName, columns, values});
    }

    /**
     * 写数据到clickhouse
     * @param dataList
     * @param baseDataTables
     * @param connection
     * @param insertSql
     */
    private void saveDefaultCommonData(List<DefaultCommonData> dataList, List<BaseDataTable> baseDataTables, ClickHouseConnection connection, String insertSql) {
        PreparedStatement preparedStatement = null;
        try {
            preparedStatement = connection.prepareStatement(insertSql);
            for (DefaultCommonData commonData : dataList) {
                preparedStatement.setLong(1, commonData.getId());
                preparedStatement.setString(2, commonData.getRecordId());
                preparedStatement.setString(3, commonData.getTypeCode());
                preparedStatement.setString(4, commonData.getYear());
                preparedStatement.setString(5, commonData.getMonth());
                preparedStatement.setString(6, commonData.getDay());
                preparedStatement.setString(7, commonData.getProvince());
                preparedStatement.setString(8, commonData.getCity());
                preparedStatement.setString(9, commonData.getRegion());
                preparedStatement.setString(10, commonData.getLng());
                preparedStatement.setString(11, commonData.getLat());
                preparedStatement.setString(12, commonData.getMultipolygon());
                Map<String, Object> indexDataMap = commonData.getIndexDataMap();
                for (int i = 13; i < baseDataTables.size() + 13; i++) {
                    preparedStatement.setString(i, MapUtil.getStr(indexDataMap, baseDataTables.get(i-13).getDataIndex(), ""));
                }
                preparedStatement.addBatch();
            }
            preparedStatement.executeBatch();
            connection.commit();
        } catch (Exception e) {
            log.error("导入数据到clickhouse失败！", e);
        } finally {
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException ignore) {
                }
            }
        }
    }

    /**
     * 解析时间
     *
     * @param record
     * @return
     */
    private BaseDataDefine.TimeWrapper buildTimeWrapper(BaseDataDefine record) {
        BaseDataDefine.TimeWrapper timeWrapper;
        //解析时间
        if ("year".equals(record.getTimeLevel())) {
            timeWrapper = new BaseDataDefine.TimeWrapper(record.getTime(), "", "");
        } else if ("month".equals(record.getTimeLevel())) {
            String[] yearMonth = StringUtils.split(record.getTime(), "-");
            if (yearMonth != null && yearMonth.length == 2) {
                timeWrapper = new BaseDataDefine.TimeWrapper(yearMonth[0], record.getTime(), "");
            } else {
                timeWrapper = new BaseDataDefine.TimeWrapper("", record.getTime(), "");
            }
        } else {
            String[] yearMonthDay = StringUtils.split(record.getTime(), "-");
            if (yearMonthDay != null && yearMonthDay.length == 3) {
                timeWrapper = new BaseDataDefine.TimeWrapper(yearMonthDay[0], yearMonthDay[0] + "-" + yearMonthDay[1], record.getTime());
            } else {
                timeWrapper = new BaseDataDefine.TimeWrapper("", "", record.getTime());
            }
        }
        return timeWrapper;
    }

    /**
     * ClickHouse建表或者修改表
     *
     * @param baseDataTables
     * @param dataRecord
     * @param dataType
     */
    private void buildTable(List<BaseDataTable> baseDataTables, BaseDataRecord dataRecord, BaseDataType dataType) {
        List<String> columns = baseDataTables.stream().map(b -> StringUtils.join("`", b.getDataIndex(), "` String")).collect(Collectors.toList());

        if (StringUtils.isBlank(dataType.getChTableName())) {
            try {
                //建表
                CurrentDataSourceContext.setDataSourceType(WdAnalysisConst.CLICKHOUSE_DATASOURCE);
                this.getBaseMapper().createClickHouseTable("`" + CH_TABLE_PREFIX + dataType.getCode() + "`", columns);
                CurrentDataSourceContext.clearDataSourceType();
                dataType.setChTableName(CH_TABLE_PREFIX + dataType.getCode());
                dataType.setChTableIndexNum(baseDataTables.size());
                baseDataTypeService.updateById(dataType);
            } catch (Exception e) {
                dataRecord.setStatus(FileParseRecordStatusEnum.IMPORT_FAIL.getStatus());
            }
        } else {
            try {
                //不用建表, 看是否需要扩充列
                if (columns.size() > dataType.getChTableIndexNum()) {
                    //ADD COLUMN
                    CurrentDataSourceContext.setDataSourceType(WdAnalysisConst.CLICKHOUSE_DATASOURCE);
                    for (int i = dataType.getChTableIndexNum(); i < columns.size(); i++) {
                        String lastCol = "`index" + i + "`";
                        this.getBaseMapper().addColumnToClickHouse(dataType.getChTableName(), "`index" + (i + 1) + "`", lastCol);
                    }
                    CurrentDataSourceContext.clearDataSourceType();
                    dataType.setChTableIndexNum(baseDataTables.size());
                    baseDataTypeService.updateById(dataType);
                }
            } catch (Exception e) {
                dataRecord.setStatus(FileParseRecordStatusEnum.IMPORT_FAIL.getStatus());
            }
        }
    }

    @Override
    public void parseFile(BaseDataRecord record) {
        record.setStatus(FileParseRecordStatusEnum.PARSING.getStatus());
        record.setDataSize(0);
        this.updateById(record);
        ThreadUtil.execute(() -> doParse(record));
    }

    @Override
    public void defineIndexData(BaseDataDefine record) {
        BaseDataRecord dataRecord = this.getById(record.getRecordId());
        dataRecord.setTypeCode("-1");
        if (ObjectUtil.isNull(dataRecord))
            throw new ServiceException(TimeSpaceDataRecordExceptionEnum.NOT_EXIST);
        this.updateById(dataRecord);
        //保存指标对应表
        titleRelevanceService.saveBatch(record.getRelevanceList());
        ThreadUtil.execute(() -> {
            // todo 解析指标数据
            biIndexDataService.addData(record);
        });
    }

    @Override
    public void parseIndexFile(BaseDataRecord record) {
        record.setStatus(FileParseRecordStatusEnum.PARSING.getStatus());
        record.setDataSize(0);
        this.updateById(record);
        ThreadUtil.execute(() -> doParse(record));
    }

    private void doParse(BaseDataRecord record) {
        FileParser fileParser = FileParserSupport.getParser(record.getExtension());
        Integer status = FileParseRecordStatusEnum.PARSED.getStatus();
        if (fileParser != null) {
            this.checkRecord(record.getId());
            ClickHouseConnection connection = null;
            ParserResult parserResult = null;
            try {
                ClickHouseDataSource clickHouseDataSource = LoadClickhouseDatasource.get();
                connection = clickHouseDataSource.getConnection();
                connection.setAutoCommit(false);
                ClickHouseConnection finalConnection = connection;
                parserResult = fileParser.parse(record.getFilePath(), 10000, record.getId().toString(), (list) -> addToClickhouse(list, finalConnection));
            } catch (Exception e) {
                log.error(record.getId() + " >>> 文件解析失败", e);
                status = FileParseRecordStatusEnum.PARSE_FAIL.getStatus();
            } finally {
                if (connection != null) {
                    try {
                        connection.close();
                    } catch (SQLException ignore) {
                    }
                }
            }
            if (parserResult != null) {
                baseDataTableService.saveByTableColumns(parserResult.getColumnList(), record.getId());
                record.setDataSize(parserResult.getTotalCount());
            }
        } else {
            log.error("没有{}文件解析器", record.getExtension());
            status = FileParseRecordStatusEnum.PARSE_FAIL.getStatus();
        }
        record.setStatus(status);
        this.updateById(record);
    }

    @Override
    public void checkRecord(Long recordId) {
        CurrentDataSourceContext.setDataSourceType(WdAnalysisConst.CLICKHOUSE_DATASOURCE);
        long count = this.getBaseMapper().selectCountByRecordId(recordId.toString());
        if (count > 0) {
            this.getBaseMapper().deleteByRecordId(recordId.toString());
        }
        CurrentDataSourceContext.clearDataSourceType();
    }

    final String insertSql = "insert into base_metadata(record_id, content, create_date) values (?, ?, ?)";

    private void addToClickhouse(List<BaseMetadata> list, ClickHouseConnection connection) {
        PreparedStatement preparedStatement = null;
        try {
            preparedStatement = connection.prepareStatement(insertSql);
            for (BaseMetadata metadata : list) {
                preparedStatement.setString(1, metadata.getRecordId());
                preparedStatement.setString(2, metadata.getContent());
                preparedStatement.setString(3, metadata.getCreateDate());
                preparedStatement.addBatch();
            }
            preparedStatement.executeBatch();
            connection.commit();
        } catch (SQLException e) {
            log.error(list.get(0).getRecordId() + ">>>写入数据库失败", e);
        } finally {
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException ignore) {
                }
            }
        }
    }


    @Override
    public PageResult<BaseDataRecord> page(TimeSpaceDataRecordParam timeSpaceDataRecordParam) {
        QueryWrapper<BaseDataRecord> queryWrapper = new QueryWrapper<>();
        if (ObjectUtil.isNotNull(timeSpaceDataRecordParam)) {

            // 根据文件名称 查询
            if (ObjectUtil.isNotEmpty(timeSpaceDataRecordParam.getTitle())) {
                queryWrapper.lambda().like(BaseDataRecord::getTitle, timeSpaceDataRecordParam.getTitle());
            }
            // 根据文件大小 查询
            if (ObjectUtil.isNotEmpty(timeSpaceDataRecordParam.getSize())) {
                queryWrapper.lambda().eq(BaseDataRecord::getSize, timeSpaceDataRecordParam.getSize());
            }
            // 根据文件大小 查询
            if (ObjectUtil.isNotEmpty(timeSpaceDataRecordParam.getSizeInfo())) {
                queryWrapper.lambda().eq(BaseDataRecord::getSizeInfo, timeSpaceDataRecordParam.getSizeInfo());
            }
            // 根据文件扩展名 查询
            if (ObjectUtil.isNotEmpty(timeSpaceDataRecordParam.getExtension())) {
                queryWrapper.lambda().eq(BaseDataRecord::getExtension, timeSpaceDataRecordParam.getExtension());
            }
            // 根据数据条数 查询
            if (ObjectUtil.isNotEmpty(timeSpaceDataRecordParam.getDataSize())) {
                queryWrapper.lambda().eq(BaseDataRecord::getDataSize, timeSpaceDataRecordParam.getDataSize());
            }
            // 根据数据类型 查询
            if (ObjectUtil.isNotEmpty(timeSpaceDataRecordParam.getTypeCode())) {
                queryWrapper.lambda().eq(BaseDataRecord::getTypeCode, timeSpaceDataRecordParam.getTypeCode());
            }
            // 根据标签 查询
            if (ObjectUtil.isNotEmpty(timeSpaceDataRecordParam.getTagId())) {
                queryWrapper.lambda().eq(BaseDataRecord::getTagId, timeSpaceDataRecordParam.getTagId());
            }
            // 根据用户id,此字段不为空表示改文件为用户经营数据 查询
            if (ObjectUtil.isNotEmpty(timeSpaceDataRecordParam.getUserId())) {
                queryWrapper.lambda().eq(BaseDataRecord::getUserId, timeSpaceDataRecordParam.getUserId());
            }
            // 根据0: 正常 1: 删除 2: 导入失败 查询
            if (ObjectUtil.isNotEmpty(timeSpaceDataRecordParam.getStatus())) {
                queryWrapper.lambda().eq(BaseDataRecord::getStatus, timeSpaceDataRecordParam.getStatus());
            }
        }
        queryWrapper.lambda().orderByDesc(BaseDataRecord::getCreateTime).ne(BaseDataRecord::getTypeCode, "-1").or((c) -> c.isNull(BaseDataRecord::getTypeCode));
        return new PageResult<>(this.page(PageFactory.defaultPage(), queryWrapper));
    }

    @Override
    public List<BaseDataRecord> list(TimeSpaceDataRecordParam timeSpaceDataRecordParam) {
        return this.list();
    }

    @Override
    public BaseDataRecord add(TimeSpaceDataRecordParam timeSpaceDataRecordParam) {
        BaseDataRecord timeSpaceDataRecord = new BaseDataRecord();
        BeanUtil.copyProperties(timeSpaceDataRecordParam, timeSpaceDataRecord);
        this.save(timeSpaceDataRecord);
        return timeSpaceDataRecord;
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void delete(List<TimeSpaceDataRecordParam> timeSpaceDataRecordParamList) {
        timeSpaceDataRecordParamList.forEach(timeSpaceDataRecordParam -> {
            this.removeById(timeSpaceDataRecordParam.getId());
        });
    }

    @Override
    public void deleteOne(Long id) {
        BaseDataRecord record = this.getById(id);
        File file = new File(record.getFilePath());
        FileUtil.del(file);
        File confFile = new File(record.getFilePath() + ".conf");
        FileUtil.del(confFile);

        this.removeById(id);
        this.deleteFromClickHouse(record.getTypeCode(), id);
    }

    @Override
    public void deleteFromClickHouse(String typeCode, Long id) {
        CurrentDataSourceContext.setDataSourceType(WdAnalysisConst.CLICKHOUSE_DATASOURCE);
        //删除base_matedata
        this.getBaseMapper().deleteByRecordId(id.toString());

        //删除其它数据 BASE_DATA_
        if (StringUtils.isNotBlank(typeCode)) {
            this.getBaseMapper().deleteBaseData(CH_TABLE_PREFIX + typeCode, id.toString());
        }
        CurrentDataSourceContext.clearDataSourceType();
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void edit(TimeSpaceDataRecordParam timeSpaceDataRecordParam) {
        BaseDataRecord timeSpaceDataRecord = this.queryTimeSpaceDataRecord(timeSpaceDataRecordParam);
        BeanUtil.copyProperties(timeSpaceDataRecordParam, timeSpaceDataRecord);
        this.updateById(timeSpaceDataRecord);
    }

    @Override
    public BaseDataRecord detail(TimeSpaceDataRecordParam timeSpaceDataRecordParam) {
        return this.queryTimeSpaceDataRecord(timeSpaceDataRecordParam);
    }

    /**
     * 获取时空数据
     *
     * @author cy
     * @date 2022-07-28 19:37:28
     */
    private BaseDataRecord queryTimeSpaceDataRecord(TimeSpaceDataRecordParam timeSpaceDataRecordParam) {
        BaseDataRecord timeSpaceDataRecord = this.getById(timeSpaceDataRecordParam.getId());
        if (ObjectUtil.isNull(timeSpaceDataRecord)) {
            throw new ServiceException(TimeSpaceDataRecordExceptionEnum.NOT_EXIST);
        }
        return timeSpaceDataRecord;
    }

    @Override
    public void export(ExportParam param) {
        //查出TypeCode对应的base_data_table表title列
        BaseDataType byCode = baseDataTypeService.getByCode(param.getTypeCode());
        if(byCode == null){
            throw new ServiceException(9,"没有找到元数据："+param.getTypeCode());
        }

        List<String> list=new ArrayList<>();
        for (int i = 1; i <= byCode.getChTableIndexNum(); i++) {
            list.add("`"+"index" + i + "`");
        }
        DataParam dataParam = new DataParam();
        if(param.getTimeType().equals("year")){
            dataParam.setYear(param.getStart());
        }else if(param.getTimeType().equals("month")){
            dataParam.setStartMonth(param.getStart());
            dataParam.setEndMonth(param.getEnd());
        }else if(param.getTimeType().equals("day")){
            dataParam.setStartDay(param.getStart());
            dataParam.setEndDay(param.getEnd());
        }else{
            throw new ServiceException(9,"getTimeType Error："+param.getTimeType());
        }
        dataParam.setProvName(param.getProvince());
        dataParam.setCityName(param.getCity());
        dataParam.setCountryName(param.getRegion());

        //开始查询clickhouse数据库
        CurrentDataSourceContext.setDataSourceType(WdAnalysisConst.CLICKHOUSE_DATASOURCE);
        List<Map<String, Object>> recordData = this.baseMapper.findRecordData(byCode.getChTableName(), list, dataParam);
        CurrentDataSourceContext.clearDataSourceType();
        JSONObject typeDATA = new JSONObject();
        if(CollectionUtil.isEmpty(recordData)){
            throw new ServiceException(9,"baseMapper.findRecordData查询失败："+byCode.getChTableName());
        }
        List<ExcelExportEntity> entityList = new ArrayList<>();
        //获取index和列的映射关系
        long recordId = Long.parseLong(String.valueOf(recordData.get(0).get("record_id")));
        List<BaseDataTable> baseDataTableList = baseDataTableService.findListByRecordId(recordId);
        if(CollectionUtil.isEmpty(baseDataTableList)){
            throw new ServiceException(9,"baseDataTableService.findListByRecordId(recordId)没有查询到记录"+recordId);
        }
        baseDataTableList.forEach(el ->{
            entityList.add(new ExcelExportEntity(el.getTitle(),el.getDataIndex()));
        });
        //使用EasyPoi导出excel
        PoiUtil.exportExcelWithStream2(param.getTypeCode()+".xls",entityList,recordData);
    }

    @Override
    @DataSource(name = WdAnalysisConst.CLICKHOUSE_DATASOURCE)
    public PageResult<BaseMetadata> pageMetadata(String recordId) {
        return new PageResult<>(this.getBaseMapper().selectMetadataPage(PageFactory.defaultPage(), recordId));
    }

    @Resource
    private BaseDataRecordCache baseDataRecordCache;
    @Override
    public JSONObject findRecordData(DataParam dataParam) {
        List<String> typeCodeList = dataParam.getTypeCodeList();
        if (CollectionUtil.isEmpty(typeCodeList)) {
            throw new ServiceException(9, "类型参数不能为空");
        }
        String[] strings = typeCodeList.toArray(new String[typeCodeList.size()]);
        dataParam.setTypeCodeList(null);
        trimLngLat(dataParam);

        JSONObject result = new JSONObject();
        for (String typeCode : strings) {
            log.info("处理基础数据开始："+typeCode);
            BaseDataType byCode = baseDataTypeService.getByCode(typeCode);
            if(byCode == null){
                continue;
            }
            JSONObject typeDATA = null;
            String key = typeCode+JSONObject.toJSONString(dataParam);
            JSONObject cacheObject = baseDataRecordCache.get(key);
            if (cacheObject != null) {
                typeDATA = cacheObject;
            }else{
                //查询clickhouse数据库
                typeDATA = findRecordData(byCode, dataParam);
                baseDataRecordCache.put(key,typeDATA);
            }

            result.put(typeCode, typeDATA);
            log.info("处理基础数据结束："+typeCode);
        }
        return result;
    }
    /**
     * 经纬度保留两位小数、缩放比例保留1为小数
     * 方便建立缓存
     * @param param
     */
    private void trimLngLat(DataParam param){
        if(org.springframework.util.StringUtils.hasText(param.getLat1())){
            if(!param.isExactLngLat()){
                param.setLat1(new BigDecimal(param.getLat1()).setScale(2,BigDecimal.ROUND_DOWN).toPlainString());
                param.setLat2(new BigDecimal(param.getLat2()).setScale(2,BigDecimal.ROUND_DOWN).toPlainString());
                param.setLng1(new BigDecimal(param.getLng1()).setScale(2,BigDecimal.ROUND_DOWN).toPlainString());
                param.setLng2(new BigDecimal(param.getLng2()).setScale(2,BigDecimal.ROUND_DOWN).toPlainString());
            }
        }
        if (param.getScale() != null) {
            double v = new BigDecimal(param.getScale()).setScale(1, BigDecimal.ROUND_HALF_UP).doubleValue();
            param.setScale(v);
        }
    }

    private JSONObject findRecordData(BaseDataType byCode,DataParam dataParam ){
        List<String> list=new ArrayList<>();
        for (int i = 1; i <= byCode.getChTableIndexNum(); i++) {
            list.add("`"+"index" + i + "`");
        }
        if (dataParam.getScale() != null) {
            if("heatmap".equals(byCode.getDisplay())){
                String sampleSQL = DistanceHelper.calcSampleSQL(dataParam.getScale(),10,20);
                dataParam.setSampleSQL(sampleSQL);
            }else{
                String sampleSQL = DistanceHelper.calcSampleSQL(dataParam.getScale(),10,14);
                dataParam.setSampleSQL(sampleSQL);
            }
        }

        String lng1 = dataParam.getLng1();
        String lng2 = dataParam.getLng2();
        String lat1 = dataParam.getLat1();
        String lat2 = dataParam.getLat2();
        if (lng1 != null && lat1 != null) {
            if(lng1.compareTo(lng2)>0){
                dataParam.setLng1(lng2);
                dataParam.setLng2(lng1);
            }
            if(lat1.compareTo(lat2)>0){
                dataParam.setLat1(lat2);
                dataParam.setLat2(lat1);
            }
        }

        JSONObject typeDATA = new JSONObject();
        CurrentDataSourceContext.setDataSourceType(WdAnalysisConst.CLICKHOUSE_DATASOURCE);
        List<Map<String, Object>> recordData = this.baseMapper.findRecordData(byCode.getChTableName(), list, dataParam);
        CurrentDataSourceContext.clearDataSourceType();

        if(CollectionUtil.isEmpty(recordData)){
            log.error("baseMapper.findRecordData查询失败："+byCode.getChTableName());
            return null;
        }

        //获取index和列的映射关系
        HashMap<String, String> columnMap = new HashMap<>();
        long recordId = Long.parseLong(String.valueOf(recordData.get(0).get("record_id")));
        List<BaseDataTable> baseDataTableList = baseDataTableService.findListByRecordId(recordId);
        if(CollectionUtil.isEmpty(baseDataTableList)){
            log.error("baseDataTableService.findListByRecordId(recordId)没有查询到记录"+recordId);
            return null;
        }
        baseDataTableList.forEach(el ->{
            columnMap.put(el.getDataIndex(), el.getTitle());
        });

        //热力图值 clickhouse表中的最后一个字段
        if("heatmap".equals(byCode.getDisplay())){
            List<Map<String, Object>> heatmapData = new ArrayList<>();
            double max = 0;
            for (Map<String, Object> m : recordData) {
                HashMap<String, Object> rowMap = new HashMap<>();
                rowMap.put("lng",m.get("lng"));
                rowMap.put("lat",m.get("lat"));
                String dataIndex = baseDataTableList.get(baseDataTableList.size() - 1).getDataIndex();
                double val = 0;
                try {
                    val = Double.parseDouble(String.valueOf(m.get(dataIndex)));
                } catch (NumberFormatException e) {
                    log.error("NumberFormatException:"+recordId+e.getMessage());
                }
                rowMap.put("count",val);
                heatmapData.add(rowMap);
                if(val>max){
                    max = val;
                }
            }
            typeDATA.put("max", max);
            recordData = heatmapData;
        }else{
            //转换列名并加入到info字段中
            for (Map<String, Object> m : recordData) {
                JSONObject info = new JSONObject();
                columnMap.keySet().forEach(k ->{
                    Object o = m.get(k);
                    m.remove(k);
                    String key_ = columnMap.get(k);
                    if("the_geom".equals(key_)){
                        return;
                    }
                    info.put(key_, o);
                });
                m.put("info", info);//将所有index数据放到info对象中
                String multipolygon = String.valueOf(m.get("multipolygon"));
                JSONArray array = biIndexDataService.changeMultipolygon(multipolygon);
                if(array.size()>0){
                    m.put("multipolygon", array);
                }
            }
        }


        typeDATA.put("data",recordData);
        typeDATA.put("name", byCode.getName());
        typeDATA.put("color", getRandomColor(byCode.getName()));
        typeDATA.put("display", byCode.getDisplay());
        return typeDATA;
    }
    public String getRandomColor(String keyword){
        Random generator = new Random(keyword.hashCode());
        int r = generator.nextInt(255);
        int g =  generator.nextInt(300 - r);
        g = g > 255 ? 255 : g;
        int b =  generator.nextInt(300 - r - g);
        b = b > 255 ? 255 : b;
        Color color = new Color(r,g,b);
        String s = "#"+Integer.toHexString(color.getRGB()).substring(2);
        return s;
    }



}
